package com.cmsr.hdpf.sink.kafka

import com.cmsr.hdpf.util.ConfigHelper
import org.apache.flink.api.common.serialization.SimpleStringSchema
import org.apache.flink.streaming.connectors.kafka.FlinkKafkaProducer

import java.util.Properties

object KafkaSink {

  def GetKafkaProducer(data: String): FlinkKafkaProducer[String] = {

    val props: Properties = new Properties()
    props.setProperty("bootstrap.servers", ConfigHelper.parse_sink_kafka_bootstrapServers)
    /*props.setProperty("client.id", scala.util.Random.nextInt(9999) + s"-${System.currentTimeMillis}")*/

    val producer = new FlinkKafkaProducer[String](
      data,  // 目标 topic
      new SimpleStringSchema(), //序列化 schema
      props  // producer 配置
    )
    producer
  }
}
